package com.atguigu.userprofile.app

import java.util.Properties

import com.atguigu.userprofile.bean.{TagInfo, TaskInfo, TaskTagRule}
import com.atguigu.userprofile.constant.ConstCode
import com.atguigu.userprofile.dao.{TagInfoDAO, TaskInfoDAO, TaskTagRuleDAO}
import com.atguigu.userprofile.util.{MyPropertiesUtil, MySqlUtil}
import org.apache.spark.SparkConf
import org.apache.spark.sql.SparkSession

/**
 * 定义一个通用的标签任务程序，用于处理 统计类标签任务。 考虑通用性，在web界面配置的标签，都能执行并将标签任务执行结果，写道具体的库中对应的标签名称的表中。
 */
object TaskTagSqlApp {


  //1、 根据TaskID 读取 任务的定义、规则、SQL    读取标签  名称
  //2、 每个标签都保存在对应的表中, 一个标签对应一张表   根据定义 建立标签表（如果新标签,需要先创建表）
  //3、 通过sql查询数据仓库中的数据 ，写入到标签表中
  //      insert overwrite  table xxx  select xxx
  //     根据定义规则进行拼接完成
  def main(args: Array[String]): Unit = {


    val sparkConf: SparkConf = new SparkConf().setAppName("task_tag_sql_app").setMaster("local[*]")
    val sparkSession: SparkSession = SparkSession.builder().config(sparkConf).enableHiveSupport().getOrCreate()


    /**
     * TODO 1、 根据TaskID 读取 任务的定义、规则、SQL    读取标签  名称
     *    1.1  taskId   //spark-submit结尾处 会带俩个参数  第一个参数是taskId  第二个参数是业务日期 ，一般是前一天的日期 . spark-submit   --master ..xxxx.xx.     xxxxx.jar  1 2021-06-08
     */

    val taskId: String = args(0)
    val taskDate: String = args(1)

    /*
      TODO 1.2  根据TaskID 读取tag_info ==> 也就是定义的标签信息,对应MySQL中 画像管理平台上的 tag_info 这张表.
        select * from tag_info where tag_task_id='1'
        TagInfo(13,Tag_Natural_Characteristic_Gender,用户性别,3,5,1,3,null,null,1,null,2022-01-19 10:00:41.0,null)
     */
    val tagInfo: TagInfo = TagInfoDAO.getTagInfoByTaskId(taskId)
    println(tagInfo)
    /*
      TODO 1.3   根据TaskID    读取task_info, 也就获取到了:标签对应的任务信息,包括:执行方式是 SparkSQL还是Spark任务.如果是SQL,SQL语句是什么,还有要提交到Spark上的任务参数.
        select id,
           task_name,
           task_status,
           task_comment,
           task_time,
           task_type,
           exec_type,
           main_class,
           file_id,
           task_args,
           task_sql,
           task_exec_level,
           create_time
        from task_info where id = 1;
        TaskInfo(1,标签计算:用户性别,2,注释,02:40,TAG,SQL,null,null,
          --driver-memory=1G
          --num-executors=3
          --executor-memory=2G
          --executor-cores=2
          --conf spark.default.parallelism=12,
        select id uid, `if`(gender is null, 'U', gender) as query_value from dim_user_info where dt = '9999-99-99',
        100,null)
     */
    val taskInfo: TaskInfo = TaskInfoDAO.getTaskInfo(taskId)
    println(taskInfo)
    /*
      TODO 1.4  根据TaskID 读取task_tag_info,也就是根据任务id,查询当前标签任务对应的 查询规则.也就是标签管理-->三级标签-->管理任务-->标签规则配置中的信息.
        List(TaskTagRule(22,13,1,M,14,男性), TaskTagRule(23,13,1,F,15,女性), TaskTagRule(24,13,1,U,16,未知))
     */
    val taskTagRuleList: List[TaskTagRule] = TaskTagRuleDAO.getTaskTagRuleListByTaskId(taskId)
    println(taskTagRuleList)


    /*
      TODO 2、 每个标签都保存在对应的表中. 一个标签对应一张表   根据定义 建立标签表（如果新标签,没有表则创建). 剩下的步骤就是:通过我们上面的查询,sql动态拼接,建表.
        create table if not exists tag_natural_characteristic_gender
          (
              uid       string,
              tag_value STRING
          )
          comment '用户性别' PARTITIONED BY (`dt` STRING)
          ROW FORMAT DELIMITED FIELDS TERMINATED BY '\t'
          LOCATION 'hdfs://hadoop102:8020/user_profile/user_profile0111/tag_natural_characteristic_gender'

     */

    val tableName = tagInfo.tagCode.toLowerCase // 获取标签的编码,通过这个创建Hive表.
    // 通过模式匹配来获取标签值类型
    val tagValueType: String = tagInfo.tagValueType match {
      case ConstCode.TAG_VALUE_TYPE_STRING => "STRING"
      case ConstCode.TAG_VALUE_TYPE_LONG => "BIGINT"
      case ConstCode.TAG_VALUE_TYPE_DECIMAL => "DECIMAL(16,2)"
      case ConstCode.TAG_VALUE_TYPE_DATE => "STRING"
    }

    val properties: Properties = MyPropertiesUtil.load("config.properties")
    /** hdfs://hadoop102:8020/user_profile */
    val hdfsPath: String = properties.getProperty("hdfs-store.path")
    /** gmall */
    val dwDbName: String = properties.getProperty("data-warehouse.dbname")
    /** user_profile0111 */
    val upDbName: String = properties.getProperty("user-profile.dbname")

    // 最终创建的表在HDFS上的位置: hdfs://hadoop102:8020/user_profile/user_profile0111/三级标签的名称
    val createSQL =
      s"""
         |create table   if not exists ${tableName} ( uid string , tag_value $tagValueType )
         |     comment '${tagInfo.tagName}' PARTITIONED BY (`dt` STRING)
         |      ROW FORMAT DELIMITED FIELDS TERMINATED BY '\\t'
         |       LOCATION    '$hdfsPath/$upDbName/$tableName'
       """.stripMargin

    println(createSQL)

    /*
      TODO 3、 通过sql查询数据仓库中的数据 ，写入到标签表中
        insert overwrite table user_profile0111.tag_natural_characteristic_gender partition (dt = '2021-06-08')
        select uid,
               case query_value
                   when 'M' then '男性'
                   when 'F' then '女性'
                   when 'U' then '未知'
                   end
                   as tag_value
        from (
                -- 定义标签任务穿过来的sql.这里通过查询Hive中的这张表,将结果返回.
                 select id uid, `if`(gender is null, 'U', gender) as query_value
                 from dim_user_info
                 where dt = '9999-99-99'
             ) tv
     */
    // 3.1  动态根据 tagRule 生成case when 语句
    // 3.2 针对sql 中的$dt要换成 业务日期
    val taskSql: String = taskInfo.taskSql.replace("$dt", taskDate)
    var caseWhenSql = ""
    /*
        判断定义的标签任务中 标签规则配置是否存在.如果我们统计的是连续值,则不会存在标签规则配置.
        eg : 通过数仓 dwt_user_topic 用户主题中,抓取最近30天用户的下单数量.sql语句如下:
          select user_id as uid,order_last_30d_count as query_value from dwt_user_topic where dt = '2020-06-14'; 其中我们传入的日期,不应该写死,因该动态传入.

     */
    if (taskTagRuleList.size > 0) {
      val whenList: List[String] = taskTagRuleList.map(taskTagRule => s"when  '${taskTagRule.queryValue}' then '${taskTagRule.subTagValue}'")
      caseWhenSql = "case  query_value " + whenList.mkString(" ") + " end  as  tag_value"
    } else {
      //如果没有子标签匹配的话 把query_value 直接作为tag_value
      caseWhenSql = "   query_value  as  tag_value"
    }

    val selectSql = s"select uid , $caseWhenSql  from (${taskSql} ) tv"

    val insertSelectSql = s"insert overwrite table   $upDbName.$tableName partition (dt='$taskDate')  $selectSql"

    println(insertSelectSql)

    /*
      TODO 在画像库中建表
       需要注意: 我们的标签任务,每次跑完后,在创建表时,如果存在会先删除再创建. ==> 考虑到的是每天需要计算的标签个数不同.
     */
    sparkSession.sql("use " + upDbName)
    sparkSession.sql(createSQL)
    // 在数仓库中执行标签任务,并将最终执行的结果,放在画像库中.
    sparkSession.sql("use " + dwDbName)
    sparkSession.sql(insertSelectSql)
  }
}
